package avicit.bdp.dds.service.process;

import avicit.bdp.common.api.TaskAPI;
import avicit.bdp.common.datasource.BaseDataSource;
import avicit.bdp.common.datasource.DataSourceFactory;
import avicit.bdp.common.dto.BdpPrmEngineResourceDTO;
import avicit.bdp.common.dto.PythonResourceDTO;
import avicit.bdp.common.dto.StorageResourceConf;
import avicit.bdp.common.service.dto.DataSourceDTO;
import avicit.bdp.common.service.service.DataSourceService;
import avicit.bdp.common.service.service.FileOperateCommonService;
import avicit.bdp.common.utils.ConfigUtils;
import avicit.bdp.common.utils.DateUtils;
import avicit.bdp.common.utils.EntityUtils;
import avicit.bdp.core.constant.ProcessConstants;
import avicit.bdp.core.util.json.JSONUtils;
import avicit.bdp.dgs.api.QualityJobApi;
import avicit.bdp.dds.api.enums.ExecuteType;
import avicit.bdp.dds.api.service.AlgoNodesService;
import avicit.bdp.dds.api.service.ExecutorService;
import avicit.bdp.dds.api.service.ProjectTableService;
import avicit.bdp.dds.api.service.TaskNodesService;
import avicit.bdp.dds.common.Constants;
import avicit.bdp.dds.common.utils.ParameterUtils;
import avicit.bdp.dds.dao.entity.AlgoNodes;
import avicit.bdp.dds.dao.entity.Command;
import avicit.bdp.dds.dao.entity.CycleDependency;
import avicit.bdp.dds.dao.entity.ErrorCommand;
import avicit.bdp.dds.dao.entity.ProcessDefinition;
import avicit.bdp.dds.dao.entity.ProcessInstance;
import avicit.bdp.dds.dao.entity.ProcessInstanceMap;
import avicit.bdp.dds.dao.entity.Project;
import avicit.bdp.dds.dao.entity.Resource;
import avicit.bdp.dds.dao.entity.Schedule;
import avicit.bdp.dds.dao.entity.TaskInstance;
import avicit.bdp.dds.dao.entity.TaskNodes;
import avicit.bdp.dds.dao.entity.UdfFunc;
import avicit.bdp.dds.dao.mapper.CommandMapper;
import avicit.bdp.dds.dao.mapper.ErrorCommandMapper;
import avicit.bdp.dds.dao.mapper.ProcessDefinitionMapper;
import avicit.bdp.dds.dao.mapper.ProcessInstanceMapMapper;
import avicit.bdp.dds.dao.mapper.ProcessInstanceMapper;
import avicit.bdp.dds.dao.mapper.ProjectMapper;
import avicit.bdp.dds.dao.mapper.ResourceMapper;
import avicit.bdp.dds.dao.mapper.ScheduleMapper;
import avicit.bdp.dds.dao.mapper.TaskInstanceMapper;
import avicit.bdp.dds.dao.mapper.UdfFuncMapper;
import avicit.bdp.dds.dispatch.enums.AuthorizationType;
import avicit.bdp.dds.dispatch.enums.CommandType;
import avicit.bdp.dds.dispatch.enums.CycleEnum;
import avicit.bdp.dds.dispatch.enums.ExecutionStatus;
import avicit.bdp.dds.dispatch.enums.FailureStrategy;
import avicit.bdp.dds.dispatch.enums.Flag;
import avicit.bdp.dds.dispatch.enums.ResourceType;
import avicit.bdp.dds.dispatch.enums.TaskDependType;
import avicit.bdp.dds.dispatch.enums.TaskType;
import avicit.bdp.dds.dispatch.enums.WarningType;
import avicit.bdp.dds.dispatch.model.DateInterval;
import avicit.bdp.dds.dispatch.model.ProcessData;
import avicit.bdp.dds.dispatch.model.TaskNode;
import avicit.bdp.dds.dispatch.process.Property;
import avicit.bdp.dds.dispatch.task.subprocess.SubProcessParameters;
import avicit.bdp.dds.remote.utils.Host;
import avicit.bdp.dds.service.log.LogClientService;
import avicit.bdp.dds.service.quartz.cron.CronUtils;
import avicit.bdp.dms.api.BdpPrmEngineResourceApi;
import avicit.platform6.commons.utils.ComUtil;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.cronutils.model.Cron;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang.ArrayUtils;
import org.apache.commons.lang3.StringUtils;
import org.quartz.CronExpression;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Calendar;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;

import static java.util.stream.Collectors.toSet;

/**
 * process relative dao that some mappers in this.
 */
@Component
public class ProcessService {

    private final Logger logger = LoggerFactory.getLogger(getClass());

    private final int[] stateArray =
            new int[]{
                    ExecutionStatus.SUBMITTED_SUCCESS.ordinal(),
                    ExecutionStatus.RUNNING_EXEUTION.ordinal(),
                    ExecutionStatus.READY_PAUSE.ordinal(),
                    ExecutionStatus.READY_STOP.ordinal()
            };

    @Autowired
    private ProcessDefinitionMapper processDefineMapper;
    @Autowired
    private ProcessInstanceMapper processInstanceMapper;
    @Autowired
    private DataSourceService dataSourceService;
    @Autowired
    private ProcessInstanceMapMapper processInstanceMapMapper;
    @Autowired
    private TaskInstanceMapper taskInstanceMapper;
    @Autowired
    private CommandMapper commandMapper;
    @Autowired
    private ScheduleMapper scheduleMapper;
    @Autowired
    private UdfFuncMapper udfFuncMapper;
    @Autowired
    private ResourceMapper resourceMapper;
    @Autowired
    private ErrorCommandMapper errorCommandMapper;
    @Autowired
    private ProjectMapper projectMapper;
    @Autowired
    private TaskNodesService taskNodesService;
    @Autowired
    private AlgoNodesService algoNodesService;
    @Autowired
    private ProjectTableService projectTableService;
    @Autowired
    private TaskAPI taskAPI;
    @Autowired
    private FileOperateCommonService fileOperateCommonService;
    @Autowired
    private ExecutorService execService;
    @Autowired
    private BdpPrmEngineResourceApi bdpPrmEngineResourceApi;

    /**
     * 数据质量对外接口
     */
    @Autowired
    private QualityJobApi qualityJobApi;

    @Value("${dcsAPI.url}")
    private String dcsAPIUrl;

    /**
     * handle Command (construct ProcessInstance from Command) , wrapped in transaction
     *
     * @param logger         logger
     * @param host           host
     * @param validThreadNum validThreadNum
     * @param command        found command
     * @return process instance
     */
    @Transactional(rollbackFor = Exception.class)
    public ProcessInstance handleCommand(
            Logger logger, String host, int validThreadNum, Command command) {
        ProcessInstance processInstance = constructProcessInstance(command, host);
        if (processInstance == null) {
            logger.error("scan command, command parameter is error: {}", command);
            moveToErrorCommand(command, "process instance is null");
            return null;
        }

        if (!checkThreadNum(command, validThreadNum)) {
            logger.info("there is not enough thread for this command: {}", command);
            return setWaitingThreadProcess(command, processInstance);
        }

        processInstance.setCommandType(command.getCommandType());
        processInstance.addHistoryCmd(CommandType.of(command.getCommandType()));
        saveProcessInstance(processInstance);

        this.setSubProcessParam(processInstance);

        delCommandById(command.getId());

        return processInstance;
    }

    /**
     * save error command, and delete original command
     *
     * @param command command
     * @param message message
     */
    @Transactional(rollbackFor = Exception.class)
    public void moveToErrorCommand(Command command, String message) {
        ErrorCommand errorCommand = new ErrorCommand(command, message);
        EntityUtils.setCreateAndUpdateInfo(errorCommand);
        this.errorCommandMapper.insert(errorCommand);
        delCommandById(command.getId());
    }

    /**
     * set process waiting thread
     *
     * @param command         command
     * @param processInstance processInstance
     * @return process instance
     */
    private ProcessInstance setWaitingThreadProcess(
            Command command, ProcessInstance processInstance) {
        processInstance.setState(ExecutionStatus.WAITTING_THREAD.getCode());
        if (command.getCommandType() != CommandType.RECOVER_WAITTING_THREAD.getCode()) {
            processInstance.addHistoryCmd(CommandType.of(command.getCommandType()));
        }
        saveProcessInstance(processInstance);
        this.setSubProcessParam(processInstance);
        createRecoveryWaitingThreadCommand(command, processInstance);
        return null;
    }

    /**
     * check thread num
     *
     * @param command        command
     * @param validThreadNum validThreadNum
     * @return if thread is enough
     */
    private boolean checkThreadNum(Command command, int validThreadNum) {
        int commandThreadCount = this.workProcessThreadNumCount(command.getProcessDefinitionId());
        return validThreadNum >= commandThreadCount;
    }

    /**
     * insert one command
     *
     * @param command command
     * @return create result
     */
    public int createCommand(Command command) {
        int result = 0;
        if (command != null) {
            if (StringUtils.isEmpty(command.getId())) {
                command.setId(ComUtil.getId());
            }
            EntityUtils.setCreateAndUpdateInfo(command);
            result = commandMapper.insert(command);
        }
        return result;
    }

    /**
     * find one command from queue list
     *
     * @return command
     */
    public Command findOneCommand() {
        return commandMapper.getOneToRun();
    }

    /**
     * check the input command exists in queue list
     *
     * @param command command
     * @return create command result
     */
    public Boolean verifyIsNeedCreateCommand(Command command) {
        boolean isNeedCreate = true;
        Map<CommandType, Integer> cmdTypeMap = new HashMap<>(16);
        cmdTypeMap.put(CommandType.REPEAT_RUNNING, 1);
        cmdTypeMap.put(CommandType.START_FAILURE_TASK_PROCESS, 1);
        cmdTypeMap.put(CommandType.RECOVER_SUSPENDED_PROCESS, 1);
        Integer commandType = command.getCommandType();

        if (cmdTypeMap.containsKey(CommandType.of(commandType))) {
            JSONObject cmdParamObj = (JSONObject) JSON.parse(command.getCommandParam());
            JSONObject tempObj;
            int processInstanceId = cmdParamObj.getInteger(Constants.CMDPARAM_RECOVER_PROCESS_ID_STRING);

            // TODO: verify
            List<Command> commands = commandMapper.selectAll();
            // for all commands
            for (Command tmpCommand : commands) {
                if (cmdTypeMap.containsKey(CommandType.of(tmpCommand.getCommandType()))) {
                    tempObj = (JSONObject) JSON.parse(tmpCommand.getCommandParam());
                    if (tempObj != null
                            && processInstanceId
                            == tempObj.getInteger(Constants.CMDPARAM_RECOVER_PROCESS_ID_STRING)) {
                        isNeedCreate = false;
                        break;
                    }
                }
            }
        }
        return isNeedCreate;
    }

    /**
     * find process instance detail by id
     *
     * @param processId processId
     * @return process instance
     */
    public ProcessInstance findProcessInstanceDetailById(String processId) {
        return processInstanceMapper.queryDetailById(processId);
    }

    /**
     * get task node list by definitionId
     *
     * @param defineId defineId
     * @return task node list
     */
    public List<TaskNode> getTaskNodeListByDefinitionId(String defineId) {
        ProcessDefinition processDefinition = processDefineMapper.selectByPrimaryKey(defineId);
        if (processDefinition == null) {
            logger.info("process define not exists");
            return null;
        }

        String processDefinitionJson = processDefinition.getProcessDefinitionJson();
        ProcessData processData = JSONUtils.parseObject(processDefinitionJson, ProcessData.class);

        // process data check
        if (null == processData) {
            logger.error("process data is null");
            return new ArrayList<>();
        }

        return processData.getTasks();
    }

    /**
     * find process instance by id
     *
     * @param processId processId
     * @return process instance
     */
    public ProcessInstance findProcessInstanceById(String processId) {
        return processInstanceMapper.selectByPrimaryKey(processId);
    }

    /**
     * find process define by id.
     *
     * @param processDefinitionId processDefinitionId
     * @return process definition
     */
    public ProcessDefinition findProcessDefineById(String processDefinitionId) {
        return processDefineMapper.selectByPrimaryKey(processDefinitionId);
    }

    /**
     * delete work process instance by id
     *
     * @param processInstanceId processInstanceId
     * @return delete process instance result
     */
    public int deleteWorkProcessInstanceById(String processInstanceId) {
        return processInstanceMapper.deleteByPrimaryKey(processInstanceId);
    }

    /**
     * delete all sub process by parent instance id
     *
     * @param processInstanceId processInstanceId
     * @return delete all sub process instance result
     */
    public int deleteAllSubWorkProcessByParentId(String processInstanceId) {
        List<String> subProcessIdList =
                processInstanceMapMapper.querySubIdListByParentId(processInstanceId);
        for (String subId : subProcessIdList) {
            deleteAllSubWorkProcessByParentId(subId);
            deleteWorkProcessMapByParentId(subId);
            removeTaskLogFile(subId);
            deleteWorkProcessInstanceById(subId);
        }
        return 1;
    }

    /**
     * remove task log file
     *
     * @param processInstanceId processInstanceId
     */
    public void removeTaskLogFile(String processInstanceId) {
        LogClientService logClient = new LogClientService();
        List<TaskInstance> taskInstanceList = findValidTaskListByProcessId(processInstanceId);
        if (CollectionUtils.isEmpty(taskInstanceList)) {
            return;
        }

        for (TaskInstance taskInstance : taskInstanceList) {
            String taskLogPath = taskInstance.getLogPath();
            if (StringUtils.isEmpty(taskInstance.getHost())) {
                continue;
            }
            int port = Constants.RPC_PORT;
            String ip;
            try {
                ip = Host.of(taskInstance.getHost()).getIp();
            } catch (Exception e) {
                // compatible old version
                ip = taskInstance.getHost();
            }
            // remove task log from logger server
            logClient.removeTaskLog(ip, port, taskLogPath);
        }
    }

    /**
     * calculate sub process number in the process define.
     *
     * @param processDefinitionId processDefinitionId
     * @return process thread num count
     */
    private Integer workProcessThreadNumCount(String processDefinitionId) {
        List<String> ids = new ArrayList<>();
        recurseFindSubProcessId(processDefinitionId, ids);
        return ids.size() + 1;
    }

    /**
     * recursive query sub process definition id by parent id.
     *
     * @param parentId parentId
     * @param ids      ids
     */
    public void recurseFindSubProcessId(String parentId, List<String> ids) {
        ProcessDefinition processDefinition = processDefineMapper.selectByPrimaryKey(parentId);
        String processDefinitionJson = processDefinition.getProcessDefinitionJson();

        ProcessData processData = JSONUtils.parseObject(processDefinitionJson, ProcessData.class);
        if (processData != null) {
            List<TaskNode> taskNodeList = processData.getTasks();
            if (taskNodeList != null && taskNodeList.size() > 0) {
                for (TaskNode taskNode : taskNodeList) {
                    String parameter = taskNode.getParams();
                    JSONObject parameterJson = JSONObject.parseObject(parameter);
                    if (parameterJson != null
                            && parameterJson.getString(Constants.CMDPARAM_SUB_PROCESS_DEFINE_ID) != null) {
                        SubProcessParameters subProcessParam =
                                JSON.parseObject(parameter, SubProcessParameters.class);
                        ids.add(subProcessParam.getProcessDefinitionId());
                        recurseFindSubProcessId(subProcessParam.getProcessDefinitionId(), ids);
                    }
                }
            }
        }
    }

    /**
     * create recovery waiting thread command when thread pool is not enough for the process instance.
     * sub work process instance need not to create recovery command. create recovery waiting thread
     * command and delete origin command at the same time. if the recovery command is exists, only
     * update the field update_time
     *
     * @param originCommand   originCommand
     * @param processInstance processInstance
     */
    public void createRecoveryWaitingThreadCommand(
            Command originCommand, ProcessInstance processInstance) {

        // sub process does not need to create wait command
        if (processInstance.getIsSubProcess() == Flag.YES.getCode()) {
            if (originCommand != null) {
                commandMapper.deleteByPrimaryKey(originCommand.getId());
            }
            return;
        }
        Map<String, String> cmdParam = new HashMap<>();
        cmdParam.put(
                Constants.CMDPARAM_RECOVERY_WAITTING_THREAD, String.valueOf(processInstance.getId()));
        // process instance quit by "waiting thread" state
        if (originCommand == null) {
            Command command = new Command();
            command.setCommandType(CommandType.RECOVER_WAITTING_THREAD.getCode());
            command.setTaskDependType(processInstance.getTaskDependType());
            command.setFailureStrategy(processInstance.getFailureStrategy());
            command.setExecutorId(processInstance.getExecutorId());
            command.setProcessDefinitionId(processInstance.getProcessDefinitionId());
            command.setCommandParam(JSONUtils.toJson(cmdParam));
            command.setWarningType(processInstance.getWarningType());
            command.setWarningUserIds(processInstance.getWarningUserIds());
            command.setScheduleTime(processInstance.getScheduleTime());
            command.setProcessInstancePriority(processInstance.getProcessInstancePriority());
            saveCommand(command);
            return;
        }

        // update the command time if current command if recover from waiting
        if (originCommand.getCommandType() == CommandType.RECOVER_WAITTING_THREAD.getCode()) {
            originCommand.setLastUpdateDate(new Date());
        } else {
            // delete old command and create new waiting thread command
            commandMapper.deleteByPrimaryKey(originCommand.getId());
            originCommand.setId("");
            originCommand.setCommandType(CommandType.RECOVER_WAITTING_THREAD.getCode());
            originCommand.setLastUpdateDate(new Date());
            originCommand.setCommandParam(JSONUtils.toJson(cmdParam));
            originCommand.setProcessInstancePriority(processInstance.getProcessInstancePriority());
        }
        saveCommand(originCommand);
    }

    /**
     * get schedule time from command
     *
     * @param command  command
     * @param cmdParam cmdParam map
     * @return date
     */
    private Date getScheduleTime(Command command, Map<String, String> cmdParam) {
        Date scheduleTime = command.getScheduleTime();
        if (scheduleTime == null) {
            if (cmdParam != null && cmdParam.containsKey(Constants.CMDPARAM_COMPLEMENT_DATA_START_DATE)) {
                scheduleTime =
                        DateUtils.stringToDate(cmdParam.get(Constants.CMDPARAM_COMPLEMENT_DATA_START_DATE));
            }
        }
        return scheduleTime;
    }

    /**
     * generate a new work process instance from command.
     *
     * @param processDefinition processDefinition
     * @param command           command
     * @param cmdParam          cmdParam map
     * @return process instance
     */
    private ProcessInstance generateNewProcessInstance(
            ProcessDefinition processDefinition, Command command, Map<String, String> cmdParam) {
        ProcessInstance processInstance = new ProcessInstance(processDefinition);
        processInstance.setState(ExecutionStatus.RUNNING_EXEUTION.getCode());
        processInstance.setRecovery(Flag.NO.getCode());
        processInstance.setStartTime(new Date());
        processInstance.setRunTimes(1);
        processInstance.setMaxTryTimes(0);
        processInstance.setProcessDefinitionId(command.getProcessDefinitionId());
        processInstance.setCommandParam(command.getCommandParam());
        processInstance.setCommandType(command.getCommandType());
        processInstance.setIsSubProcess(Flag.NO.getCode());
        processInstance.setTaskDependType(command.getTaskDependType());
        processInstance.setFailureStrategy(command.getFailureStrategy());
        processInstance.setExecutorId(command.getExecutorId());
        Integer warningType =
                command.getWarningType() == null ? WarningType.NONE.getCode() : command.getWarningType();
        processInstance.setWarningType(warningType);
        processInstance.setWarningUserIds(command.getWarningUserIds());
        // processInstance.setWarningGroupId(warningGroupId);

        // schedule time
        Date scheduleTime = getScheduleTime(command, cmdParam);
        if (scheduleTime != null) {
            processInstance.setScheduleTime(scheduleTime);
        }
        processInstance.setCommandStartTime(command.getStartTime());
        processInstance.setLocations(processDefinition.getLocations());
        processInstance.setConnects(processDefinition.getConnects());
        // curing global params
        processInstance.setGlobalParams(
                ParameterUtils.curingGlobalParams(
                        processDefinition.getGlobalParamMap(),
                        processDefinition.getGlobalParamList(),
                        getCommandTypeIfComplement(processInstance, command),
                        processInstance.getScheduleTime()));

        // copy process define json to process instance
        processInstance.setProcessInstanceJson(processDefinition.getProcessDefinitionJson());
        // set process instance priority
        processInstance.setProcessInstancePriority(command.getProcessInstancePriority());
        String workerGroup =
                StringUtils.isBlank(command.getWorkerGroup())
                        ? Constants.DEFAULT_WORKER_GROUP
                        : command.getWorkerGroup();
        processInstance.setWorkerGroup(workerGroup);
        // processInstance.setTimeout(processDefinition.getTimeout());
        // processInstance.setTenantId(processDefinition.getTenantId());
        return processInstance;
    }

    /**
     * get process tenant there is tenant id in definition, use the tenant of the definition. if there
     * is not tenant id in the definiton or the tenant not exist use definition creator's tenant.
     *
     * @return tenant
     */
    public String getTenantForProcess() {
        return ConfigUtils.getInstance().getString("tentant", "root");
    }

    public String getQueueForProcess() {
        return ConfigUtils.getInstance().getString("queue", "default");
    }

    /**
     * check command parameters is valid
     *
     * @param command  command
     * @param cmdParam cmdParam map
     * @return whether command param is valid
     */
    private Boolean checkCmdParam(Command command, Map<String, String> cmdParam) {
        if (command.getTaskDependType() == null) {
            return true;
        }

        if (command.getTaskDependType() == TaskDependType.TASK_ONLY.getCode()
                || command.getTaskDependType() == TaskDependType.TASK_PRE.getCode()) {
            if (cmdParam == null
                    || !cmdParam.containsKey(Constants.CMDPARAM_START_NODE_NAMES)
                    || cmdParam.get(Constants.CMDPARAM_START_NODE_NAMES).isEmpty()) {
                logger.error(
                        "command node depend type is {}, but start nodes is null ",
                        command.getTaskDependType());
                return false;
            }
        }
        return true;
    }

    /**
     * 查询bdp_process_instance表，根据processDefinitionId判断是否有正在运行的ProcessInstance 如果有，禁止创建任务实例；否则允许创建
     *
     * @param processDefinitionId process definition Id
     * @return 判断是否有正在运行的ProcessInstance
     */
    private boolean checkIncompatibleProcessInstance(String processDefinitionId) {
        ProcessInstance processInstance =
                processInstanceMapper.queryRunningProcess(
                        processDefinitionId, ExecutionStatus.RUNNING_EXEUTION);
        if (processInstance != null) {
            logger.warn(
                    "Process definition {} has a instance running, can't create a new instance.",
                    processDefinitionId);
            return false;
        }

        return true;
    }

    /**
     * construct process instance according to one command.
     *
     * @param command command
     * @param host    host
     * @return process instance
     */
    private ProcessInstance constructProcessInstance(Command command, String host) {
        ProcessInstance processInstance;
        ProcessDefinition processDefinition = null;
        if (StringUtils.isNotEmpty(command.getProcessDefinitionId())) {
            processDefinition = processDefineMapper.selectByPrimaryKey(command.getProcessDefinitionId());
            if (processDefinition == null) {
                logger.error(
                        "cannot find the work process define! define id : {}",
                        command.getProcessDefinitionId());
                return null;
            }
            // 任务互斥
            if (!checkIncompatibleProcessInstance(command.getProcessDefinitionId())) {
                return null;
            }
        }

        Map<String, String> cmdParam = JSONUtils.toMap(command.getCommandParam());
        if (cmdParam != null) {
            String processInstanceId = "";
            // recover from failure or pause tasks
            if (cmdParam.containsKey(Constants.CMDPARAM_RECOVER_PROCESS_ID_STRING)) {
                String processId = cmdParam.get(Constants.CMDPARAM_RECOVER_PROCESS_ID_STRING);
                processInstanceId = processId;
                if (StringUtils.isEmpty(processInstanceId)) {
                    logger.error("command parameter is error, [ ProcessInstanceId ] is 0");
                    return null;
                }
            } else if (cmdParam.containsKey(Constants.CMDPARAM_SUB_PROCESS)) {
                // sub process map
                String pId = cmdParam.get(Constants.CMDPARAM_SUB_PROCESS);
                processInstanceId = pId;
            } else if (cmdParam.containsKey(Constants.CMDPARAM_RECOVERY_WAITTING_THREAD)) {
                // waiting thread command
                String pId = cmdParam.get(Constants.CMDPARAM_RECOVERY_WAITTING_THREAD);
                processInstanceId = pId;
            }
            if (StringUtils.isEmpty(processInstanceId)) {
                processInstance = generateNewProcessInstance(processDefinition, command, cmdParam);
            } else {
                processInstance = this.findProcessInstanceDetailById(processInstanceId);
            }

            // reset command parameter
            if (processInstance.getCommandParam() != null) {
                Map<String, String> processCmdParam = JSONUtils.toMap(processInstance.getCommandParam());
                for (Map.Entry<String, String> entry : processCmdParam.entrySet()) {
                    if (!cmdParam.containsKey(entry.getKey())) {
                        cmdParam.put(entry.getKey(), entry.getValue());
                    }
                }
            }
            // reset command parameter if sub process
            if (cmdParam.containsKey(Constants.CMDPARAM_SUB_PROCESS)) {
                processInstance.setCommandParam(command.getCommandParam());
            }
        } else {
            // generate one new process instance
            processInstance = generateNewProcessInstance(processDefinition, command, cmdParam);
        }
        if (!checkCmdParam(command, cmdParam)) {
            logger.error("command parameter check failed!");
            return null;
        }

        if (command.getScheduleTime() != null) {
            processInstance.setScheduleTime(command.getScheduleTime());
        }

        processInstance.setHost(host);

        if (processDefinition != null) {
            // 设置流程实例任务类型
            processInstance.setType(processDefinition.getType());
            // 设置流程实例附加参数，比如存放所有任务相关的时间参数
            processInstance.setExtraParams(processDefinition.getExtraParams());

            processInstance.setProcessDefinition(processDefinition);
        }

        Integer runStatus = ExecutionStatus.RUNNING_EXEUTION.getCode();
        int runTime = processInstance.getRunTimes();

        CommandType commandType = CommandType.of(command.getCommandType());
        switch (commandType) {
            case START_PROCESS:
                break;
            case START_FAILURE_TASK_PROCESS:
                // find failed tasks and init these tasks
                List<String> failedList =
                        this.findTaskIdByInstanceState(processInstance.getId(), ExecutionStatus.FAILURE);
                List<String> toleranceList =
                        this.findTaskIdByInstanceState(
                                processInstance.getId(), ExecutionStatus.NEED_FAULT_TOLERANCE);
                List<String> killedList =
                        this.findTaskIdByInstanceState(processInstance.getId(), ExecutionStatus.KILL);

                cmdParam.remove(Constants.CMDPARAM_RECOVERY_START_NODE_STRING);

                failedList.addAll(killedList);
                failedList.addAll(toleranceList);
                for (String taskId : failedList) {
                    initTaskInstance(this.findTaskInstanceById(taskId));
                }
                cmdParam.put(
                        Constants.CMDPARAM_RECOVERY_START_NODE_STRING,
                        String.join(Constants.COMMA, failedList));
                processInstance.setCommandParam(JSONUtils.toJson(cmdParam));
                processInstance.setRunTimes(runTime + 1);
                break;
            case START_CURRENT_TASK_PROCESS:
                break;
            case RECOVER_WAITTING_THREAD:
                break;
            case RECOVER_SUSPENDED_PROCESS:
                // find pause tasks and init task's state
                cmdParam.remove(Constants.CMDPARAM_RECOVERY_START_NODE_STRING);
                List<String> suspendedNodeList =
                        this.findTaskIdByInstanceState(processInstance.getId(), ExecutionStatus.PAUSE);
                List<String> stopNodeList =
                        findTaskIdByInstanceState(processInstance.getId(), ExecutionStatus.KILL);
                suspendedNodeList.addAll(stopNodeList);
                for (String taskId : suspendedNodeList) {
                    // initialize the pause state
                    initTaskInstance(this.findTaskInstanceById(taskId));
                }
                cmdParam.put(
                        Constants.CMDPARAM_RECOVERY_START_NODE_STRING, String.join(",", suspendedNodeList));
                processInstance.setCommandParam(JSONUtils.toJson(cmdParam));
                processInstance.setRunTimes(runTime + 1);
                break;
            case RECOVER_TOLERANCE_FAULT_PROCESS:
                // recover tolerance fault process
                processInstance.setRecovery(Flag.YES.getCode());
                runStatus = processInstance.getState();
                break;
            case COMPLEMENT_DATA:
                // delete all the valid tasks when complement data
                List<TaskInstance> taskInstanceList =
                        this.findValidTaskListByProcessId(processInstance.getId());
                for (TaskInstance taskInstance : taskInstanceList) {
                    taskInstance.setFlag(Flag.NO.getCode());
                    this.updateTaskInstance(taskInstance);
                }
                initComplementDataParam(processDefinition, processInstance, cmdParam);
                break;
            case REPEAT_RUNNING:
                // delete the recover task names from command parameter
                if (cmdParam != null
                        && cmdParam.containsKey(Constants.CMDPARAM_RECOVERY_START_NODE_STRING)) {
                    cmdParam.remove(Constants.CMDPARAM_RECOVERY_START_NODE_STRING);
                    processInstance.setCommandParam(JSONUtils.toJson(cmdParam));
                }
                // delete all the valid tasks when repeat running
                List<TaskInstance> validTaskList = findValidTaskListByProcessId(processInstance.getId());
                for (TaskInstance taskInstance : validTaskList) {
                    taskInstance.setFlag(Flag.NO.getCode());
                    updateTaskInstance(taskInstance);
                }
                processInstance.setStartTime(new Date());
                processInstance.setEndTime(null);
                processInstance.setRunTimes(runTime + 1);
                initComplementDataParam(processDefinition, processInstance, cmdParam);
                break;
            case SCHEDULER:
                break;
            default:
                break;
        }

        processInstance.setState(runStatus);
        return processInstance;
    }

    /**
     * return complement data if the process start with complement data
     *
     * @param processInstance processInstance
     * @param command         command
     * @return command type
     */
    private Integer getCommandTypeIfComplement(ProcessInstance processInstance, Command command) {
        if (CommandType.COMPLEMENT_DATA.getCode() == processInstance.getCmdTypeIfComplement()) {
            return CommandType.COMPLEMENT_DATA.getCode();
        } else {
            return command.getCommandType();
        }
    }

    /**
     * initialize complement data parameters
     *
     * @param processDefinition processDefinition
     * @param processInstance   processInstance
     * @param cmdParam          cmdParam
     */
    private void initComplementDataParam(
            ProcessDefinition processDefinition,
            ProcessInstance processInstance,
            Map<String, String> cmdParam) {
        if (!processInstance.isComplementData()) {
            return;
        }

        Date startComplementTime =
                DateUtils.parse(
                        cmdParam.get(Constants.CMDPARAM_COMPLEMENT_DATA_START_DATE),
                        avicit.bdp.core.constant.Constants.YYYY_MM_DD_HH_MM_SS);
        if (Flag.NO.getCode() == processInstance.getIsSubProcess()) {
            processInstance.setScheduleTime(startComplementTime);
        }
        processInstance.setGlobalParams(
                ParameterUtils.curingGlobalParams(
                        processDefinition.getGlobalParamMap(),
                        processDefinition.getGlobalParamList(),
                        CommandType.COMPLEMENT_DATA.getCode(),
                        processInstance.getScheduleTime()));
    }

    /**
     * set sub work process parameters. handle sub work process instance, update relation table and
     * command parameters set sub work process flag, extends parent work process command parameters
     *
     * @param subProcessInstance subProcessInstance
     * @return process instance
     */
    public ProcessInstance setSubProcessParam(ProcessInstance subProcessInstance) {
        String cmdParam = subProcessInstance.getCommandParam();
        if (StringUtils.isEmpty(cmdParam)) {
            return subProcessInstance;
        }
        Map<String, String> paramMap = JSONUtils.toMap(cmdParam);
        // write sub process id into cmd param.
        if (paramMap != null
                && paramMap.containsKey(Constants.CMDPARAM_SUB_PROCESS)
                && Constants.CMDPARAM_EMPTY_SUB_PROCESS.equals(
                paramMap.get(Constants.CMDPARAM_SUB_PROCESS))) {
            paramMap.remove(Constants.CMDPARAM_SUB_PROCESS);
            paramMap.put(Constants.CMDPARAM_SUB_PROCESS, String.valueOf(subProcessInstance.getId()));
            subProcessInstance.setCommandParam(JSONUtils.toJson(paramMap));
            subProcessInstance.setIsSubProcess(Flag.YES.getCode());
            this.saveProcessInstance(subProcessInstance);
        }

        if (paramMap != null) {
            // copy parent instance user def params to sub process..
            String parentInstanceId = paramMap.get(Constants.CMDPARAM_SUB_PROCESS_PARENT_INSTANCE_ID);
            if (StringUtils.isNotEmpty(parentInstanceId)) {
                ProcessInstance parentInstance = findProcessInstanceDetailById(parentInstanceId);
                if (parentInstance != null) {
                    subProcessInstance.setGlobalParams(
                            joinGlobalParams(
                                    parentInstance.getGlobalParams(), subProcessInstance.getGlobalParams()));
                    this.saveProcessInstance(subProcessInstance);
                } else {
                    logger.error(
                            "sub process command params error, cannot find parent instance: {} ", cmdParam);
                }
            }
        }
        ProcessInstanceMap processInstanceMap =
                JSONUtils.parseObject(cmdParam, ProcessInstanceMap.class);
        if (processInstanceMap == null
                || StringUtils.isEmpty(processInstanceMap.getParentProcessInstanceId())) {
            return subProcessInstance;
        }
        // update sub process id to process map table
        processInstanceMap.setProcessInstanceId(subProcessInstance.getId());

        this.updateWorkProcessInstanceMap(processInstanceMap);
        return subProcessInstance;
    }

    /**
     * join parent global params into sub process. only the keys doesn't in sub process global would
     * be joined.
     *
     * @param parentGlobalParams parentGlobalParams
     * @param subGlobalParams    subGlobalParams
     * @return global params join
     */
    private String joinGlobalParams(String parentGlobalParams, String subGlobalParams) {
        List<Property> parentPropertyList = JSONUtils.toList(parentGlobalParams, Property.class);
        List<Property> subPropertyList = JSONUtils.toList(subGlobalParams, Property.class);
        Map<String, String> subMap =
                subPropertyList.stream().collect(Collectors.toMap(Property::getProp, Property::getValue));

        for (Property parent : parentPropertyList) {
            if (!subMap.containsKey(parent.getProp())) {
                subPropertyList.add(parent);
            }
        }
        return JSONUtils.toJson(subPropertyList);
    }

    /**
     * initialize task instance
     *
     * @param taskInstance taskInstance
     */
    private void initTaskInstance(TaskInstance taskInstance) {

        if (!taskInstance.isSubProcess()) {
            ExecutionStatus state = ExecutionStatus.of(taskInstance.getState());
            if (state.typeIsCancel() || state.typeIsFailure()) {
                taskInstance.setFlag(Flag.NO.getCode());
                updateTaskInstance(taskInstance);
                return;
            }
        }
        taskInstance.setState(ExecutionStatus.SUBMITTED_SUCCESS.getCode());
        updateTaskInstance(taskInstance);
    }

    /**
     * submit task to db submit sub process to command
     *
     * @param taskInstance taskInstance
     * @return task instance
     */
    @Transactional(rollbackFor = Exception.class)
    public TaskInstance submitTask(TaskInstance taskInstance) {
        ProcessInstance processInstance =
                this.findProcessInstanceDetailById(taskInstance.getProcessInstanceId());
        logger.info(
                "start submit task : {}, instance id:{}, state: {}",
                taskInstance.getName(),
                taskInstance.getProcessInstanceId(),
                processInstance.getState());
        // submit to db
        TaskInstance task = submitTaskInstanceToDB(taskInstance, processInstance);
        if (task == null) {
            logger.error(
                    "end submit task to db error, task name:{}, process id:{} state: {} ",
                    taskInstance.getName(),
                    taskInstance.getProcessInstance(),
                    processInstance.getState());
            return null;
        }
        if (!ExecutionStatus.of(task.getState()).typeIsFinished()) {
            createSubWorkProcessCommand(processInstance, task);
        }

        logger.info(
                "end submit task to db successfully:{} state:{} complete, instance id:{} state: {}  ",
                taskInstance.getName(),
                task.getState(),
                processInstance.getId(),
                processInstance.getState());
        return task;
    }

    /**
     * set work process instance map
     *
     * @param parentInstance parentInstance
     * @param parentTask     parentTask
     * @return process instance map
     */
    private ProcessInstanceMap setProcessInstanceMap(
            ProcessInstance parentInstance, TaskInstance parentTask) {
        ProcessInstanceMap processMap =
                findWorkProcessMapByParent(parentInstance.getId(), parentTask.getId());
        if (processMap != null) {
            return processMap;
        } else if (parentInstance.getCommandType() == CommandType.REPEAT_RUNNING.getCode()
                || parentInstance.isComplementData()) {
            // update current task id to map
            // repeat running  does not generate new sub process instance
            processMap = findPreviousTaskProcessMap(parentInstance, parentTask);
            if (processMap != null) {
                processMap.setParentTaskInstanceId(parentTask.getId());
                updateWorkProcessInstanceMap(processMap);
                return processMap;
            }
        }
        // new task
        processMap = new ProcessInstanceMap();
        processMap.setId(ComUtil.getId());
        processMap.setParentProcessInstanceId(parentInstance.getId());
        processMap.setParentTaskInstanceId(parentTask.getId());
        createWorkProcessInstanceMap(processMap);
        return processMap;
    }

    /**
     * find previous task work process map.
     *
     * @param parentProcessInstance parentProcessInstance
     * @param parentTask            parentTask
     * @return process instance map
     */
    private ProcessInstanceMap findPreviousTaskProcessMap(
            ProcessInstance parentProcessInstance, TaskInstance parentTask) {
        String preTaskId;
        List<TaskInstance> preTaskList =
                this.findPreviousTaskListByWorkProcessId(parentProcessInstance.getId());
        for (TaskInstance task : preTaskList) {
            if (task.getName().equals(parentTask.getName())) {
                preTaskId = task.getId();
                ProcessInstanceMap map =
                        findWorkProcessMapByParent(parentProcessInstance.getId(), preTaskId);
                if (map != null) {
                    return map;
                }
            }
        }
        logger.info(
                "sub process instance is not found,parent task:{},parent instance:{}",
                parentTask.getId(),
                parentProcessInstance.getId());
        return null;
    }

    /**
     * create sub work process command
     *
     * @param parentProcessInstance parentProcessInstance
     * @param task                  task
     */
    private void createSubWorkProcessCommand(
            ProcessInstance parentProcessInstance, TaskInstance task) {
        if (!task.isSubProcess()) {
            return;
        }
        ProcessInstanceMap instanceMap = setProcessInstanceMap(parentProcessInstance, task);
        TaskNode taskNode = JSONUtils.parseObject(task.getTaskJson(), TaskNode.class);
        Map<String, String> subProcessParam = JSONUtils.toMap(taskNode.getParams());
        String childDefineId = subProcessParam.get(Constants.CMDPARAM_SUB_PROCESS_DEFINE_ID);

        ProcessInstance childInstance =
                findSubProcessInstance(parentProcessInstance.getId(), task.getId());

        Integer fatherType = parentProcessInstance.getCommandType();
        Integer commandType = fatherType;
        if (childInstance == null) {
            String fatherHistoryCommand = parentProcessInstance.getHistoryCmd();
            // sub process must begin with schedule/complement data
            // if father begin with scheduler/complement data
            if (fatherHistoryCommand.startsWith(CommandType.SCHEDULER.toString())
                    || fatherHistoryCommand.startsWith(CommandType.COMPLEMENT_DATA.toString())) {
                commandType = CommandType.valueOf(fatherHistoryCommand.split(Constants.COMMA)[0]).getCode();
            }
        }

        if (childInstance != null) {
            childInstance.setState(ExecutionStatus.SUBMITTED_SUCCESS.getCode());
            updateProcessInstance(childInstance);
        }
        // set sub work process command
        String processMapStr = JSONUtils.toJson(instanceMap);
        Map<String, String> cmdParam = JSONUtils.toMap(processMapStr);

        if (commandType == CommandType.COMPLEMENT_DATA.getCode()
                || (childInstance != null && childInstance.isComplementData())) {
            Map<String, String> parentParam = JSONUtils.toMap(parentProcessInstance.getCommandParam());
            String endTime = parentParam.get(Constants.CMDPARAM_COMPLEMENT_DATA_END_DATE);
            String startTime = parentParam.get(Constants.CMDPARAM_COMPLEMENT_DATA_START_DATE);
            cmdParam.put(Constants.CMDPARAM_COMPLEMENT_DATA_END_DATE, endTime);
            cmdParam.put(Constants.CMDPARAM_COMPLEMENT_DATA_START_DATE, startTime);
            processMapStr = JSONUtils.toJson(cmdParam);
        }

        updateSubProcessDefinitionByParent(parentProcessInstance, childDefineId);

        Command command = new Command();
        command.setWarningType(parentProcessInstance.getWarningType());
        command.setWarningUserIds(parentProcessInstance.getWarningUserIds());
        command.setFailureStrategy(parentProcessInstance.getFailureStrategy());
        command.setProcessDefinitionId(childDefineId);
        command.setScheduleTime(parentProcessInstance.getScheduleTime());
        command.setExecutorId(parentProcessInstance.getExecutorId());
        command.setCommandParam(processMapStr);
        command.setCommandType(commandType);
        command.setProcessInstancePriority(parentProcessInstance.getProcessInstancePriority());
        command.setWorkerGroup(parentProcessInstance.getWorkerGroup());
        command.setTaskDependType(TaskDependType.TASK_POST.getCode()); // 执行所有节点

        int result = createCommand(command);
        if (result > 0) {
            logger.info("sub process command created: {} ", command.toString());
        } else {
            logger.error("create process command error!");
        }
    }

    /**
     * update sub process definition
     *
     * @param parentProcessInstance parentProcessInstance
     * @param childDefinitionId     childDefinitionId
     */
    private void updateSubProcessDefinitionByParent(
            ProcessInstance parentProcessInstance, String childDefinitionId) {
        ProcessDefinition fatherDefinition =
                this.findProcessDefineById(parentProcessInstance.getProcessDefinitionId());
        ProcessDefinition childDefinition = this.findProcessDefineById(childDefinitionId);
        if (childDefinition != null && fatherDefinition != null) {
            // childDefinition.setReceivers(fatherDefinition.getReceivers());
            // childDefinition.setReceiversCc(fatherDefinition.getReceiversCc());
            processDefineMapper.updateByPrimaryKey(childDefinition);
        }
    }

    /**
     * submit task to mysql
     *
     * @param taskInstance    taskInstance
     * @param processInstance processInstance
     * @return task instance
     */
    public TaskInstance submitTaskInstanceToDB(
            TaskInstance taskInstance, ProcessInstance processInstance) {
        Integer processInstanceState = processInstance.getState();

        if (ExecutionStatus.of(taskInstance.getState()).typeIsFailure()) {
            if (taskInstance.isSubProcess()) {
                taskInstance.setRetryTimes(taskInstance.getRetryTimes() + 1);
            } else {

                if (processInstanceState != ExecutionStatus.READY_STOP.getCode()
                        && processInstanceState != ExecutionStatus.READY_PAUSE.getCode()) {
                    // failure task set invalid
                    taskInstance.setFlag(Flag.NO.getCode());
                    updateTaskInstance(taskInstance);
                    // crate new task instance
                    if (taskInstance.getState() != ExecutionStatus.NEED_FAULT_TOLERANCE.getCode()) {
                        taskInstance.setRetryTimes(taskInstance.getRetryTimes() + 1);
                    }
                    taskInstance.setEndTime(null);
                    taskInstance.setStartTime(new Date());
                    taskInstance.setFlag(Flag.YES.getCode());
                    taskInstance.setHost(null);
                    taskInstance.setId("");
                }
            }
        }
        taskInstance.setExecutorId(processInstance.getExecutorId());
        taskInstance.setProcessInstancePriority(processInstance.getProcessInstancePriority());
        taskInstance.setState(getSubmitTaskState(taskInstance, processInstanceState));
        taskInstance.setSubmitTime(new Date());
        boolean saveResult = saveTaskInstance(taskInstance);
        if (!saveResult) {
            return null;
        }
        return taskInstance;
    }

    /**
     * ${processInstancePriority}_${processInstanceId}_${taskInstancePriority}_${taskInstanceId}_${task
     * executed by ip1},${ip2}... The tasks with the highest priority are selected by comparing the
     * priorities of the above four levels from high to low.
     *
     * @param taskInstance taskInstance
     * @return task zk queue str
     */
    public String taskZkInfo(TaskInstance taskInstance) {

        // String taskWorkerGroup = getTaskWorkerGroup(taskInstance);
        ProcessInstance processInstance =
                this.findProcessInstanceById(taskInstance.getProcessInstanceId());
        if (processInstance == null) {
            logger.error(
                    "process instance is null. please check the task info, task id: " + taskInstance.getId());
            return "";
        }

        return processInstance.getProcessInstancePriority()
                + Constants.UNDERLINE
                + taskInstance.getProcessInstanceId()
                + Constants.UNDERLINE
                + taskInstance.getTaskInstancePriority()
                + Constants.UNDERLINE
                + taskInstance.getId()
                + Constants.UNDERLINE
                + taskInstance.getWorkerGroup();
    }

    /**
     * get submit task instance state by the work process state cannot modify the task state when
     * running/kill/submit success, or this task instance is already exists in task queue . return
     * pause if work process state is ready pause return stop if work process state is ready stop if
     * all of above are not satisfied, return submit success
     *
     * @param taskInstance         taskInstance
     * @param processInstanceState processInstanceState
     * @return process instance state
     */
    public Integer getSubmitTaskState(TaskInstance taskInstance, Integer processInstanceState) {
        Integer state = taskInstance.getState();
        // running or killed
        // the task already exists in task queue
        // return state
        if (state == ExecutionStatus.RUNNING_EXEUTION.getCode()
                || state == ExecutionStatus.KILL.getCode()
                || checkTaskExistsInTaskQueue(taskInstance)) {
            return state;
        }
        // return pause /stop if process instance state is ready pause / stop
        // or return submit success
        if (processInstanceState == ExecutionStatus.READY_PAUSE.getCode()) {
            state = ExecutionStatus.PAUSE.getCode();
        } else if (processInstanceState == ExecutionStatus.READY_STOP.getCode()
                || !checkProcessStrategy(taskInstance)) {
            state = ExecutionStatus.KILL.getCode();
        } else {
            state = ExecutionStatus.SUBMITTED_SUCCESS.getCode();
        }
        return state;
    }

    /**
     * check process instance strategy
     *
     * @param taskInstance taskInstance
     * @return check strategy result
     */
    private boolean checkProcessStrategy(TaskInstance taskInstance) {
        ProcessInstance processInstance =
                this.findProcessInstanceById(taskInstance.getProcessInstanceId());
        Integer failureStrategy = processInstance.getFailureStrategy();
        if (failureStrategy == FailureStrategy.CONTINUE.getCode()) {
            return true;
        }
        List<TaskInstance> taskInstances =
                this.findValidTaskListByProcessId(taskInstance.getProcessInstanceId());

        for (TaskInstance task : taskInstances) {
            if (task.getState() == ExecutionStatus.FAILURE.getCode()) {
                return false;
            }
        }
        return true;
    }

    /**
     * check the task instance existing in queue
     *
     * @param taskInstance taskInstance
     * @return whether taskinstance exists queue
     */
    public boolean checkTaskExistsInTaskQueue(TaskInstance taskInstance) {
        if (taskInstance.isSubProcess()) {
            return false;
        }

        String taskZkInfo = taskZkInfo(taskInstance);
        return false;
    }

    /**
     * create a new process instance
     *
     * @param processInstance processInstance
     */
    public void createProcessInstance(ProcessInstance processInstance) {
        if (processInstance != null) {
            processInstanceMapper.insert(processInstance);
        }
    }

    /**
     * insert or update work process instance to data base
     *
     * @param processInstance processInstance
     */
    public void saveProcessInstance(ProcessInstance processInstance) {
        if (processInstance == null) {
            logger.error("save error, process instance is null!");
            return;
        }
        if (StringUtils.isNotEmpty(processInstance.getId())) {
            processInstanceMapper.updateByPrimaryKey(processInstance);
        } else {
            processInstance.setId(ComUtil.getId());
            createProcessInstance(processInstance);
        }
    }

    /**
     * insert or update command
     *
     * @param command command
     * @return save command result
     */
    public int saveCommand(Command command) {
        if (command != null && command.getStartTime() == null) {
            command.setStartTime(new Date());
        }
        EntityUtils.setCreateAndUpdateInfo(command);
        if (StringUtils.isNotBlank(command.getId())) {
            return commandMapper.updateByPrimaryKey(command);
        } else {
            command.setId(ComUtil.getId());
            return commandMapper.insert(command);
        }
    }

    /**
     * insert or update task instance
     *
     * @param taskInstance taskInstance
     * @return save task instance result
     */
    public boolean saveTaskInstance(TaskInstance taskInstance) {
        if (StringUtils.isNotEmpty(taskInstance.getId())) {
            return updateTaskInstance(taskInstance);
        } else {
            String taskInstanceId = ComUtil.getId();
            taskInstance.setId(taskInstanceId);
            return createTaskInstance(taskInstance);
        }
    }

    /**
     * insert task instance
     *
     * @param taskInstance taskInstance
     * @return create task instance result
     */
    public boolean createTaskInstance(TaskInstance taskInstance) {
        int count = taskInstanceMapper.insert(taskInstance);
        return count > 0;
    }

    /**
     * update task instance
     *
     * @param taskInstance taskInstance
     * @return update task instance result
     */
    public boolean updateTaskInstance(TaskInstance taskInstance) {
        int count = taskInstanceMapper.updateByPrimaryKey(taskInstance);
        return count > 0;
    }

    public boolean updateTaskInstanceByPrimaryKeySelective(TaskInstance taskInstance) {
        int count = taskInstanceMapper.updateByPrimaryKeySelective(taskInstance);
        return count > 0;
    }

    /**
     * delete a command by id
     *
     * @param id id
     */
    public void delCommandById(String id) {
        commandMapper.deleteByPrimaryKey(id);
    }

    /**
     * find task instance by id
     *
     * @param taskId task id
     * @return task intance
     */
    public TaskInstance findTaskInstanceById(String taskId) {
        return taskInstanceMapper.selectByPrimaryKey(taskId);
    }

    /**
     * package task instance，associate processInstance and processDefine
     *
     * @param taskInstId taskInstId
     * @return task instance
     */
    public TaskInstance getTaskInstanceDetailByTaskId(String taskInstId) {
        // get task instance
        TaskInstance taskInstance = findTaskInstanceById(taskInstId);
        if (taskInstance == null) {
            return null;
        }
        // get process instance
        ProcessInstance processInstance =
                findProcessInstanceDetailById(taskInstance.getProcessInstanceId());
        // get process define
        ProcessDefinition processDefine = findProcessDefineById(taskInstance.getProcessDefinitionId());

        taskInstance.setProcessInstance(processInstance);
        taskInstance.setProcessDefine(processDefine);
        return taskInstance;
    }

    /**
     * get id list by task state
     *
     * @param instanceId instanceId
     * @param state      state
     * @return task instance states
     */
    public List<String> findTaskIdByInstanceState(String instanceId, ExecutionStatus state) {
        return taskInstanceMapper.queryTaskByProcessIdAndState(instanceId, state.ordinal());
    }

    /**
     * find valid task list by process definition id
     *
     * @param processInstanceId processInstanceId
     * @return task instance list
     */
    public List<TaskInstance> findValidTaskListByProcessId(String processInstanceId) {
        return taskInstanceMapper.findValidTaskListByProcessId(processInstanceId, Flag.YES.getCode());
    }

    /**
     * find previous task list by work process id
     *
     * @param processInstanceId processInstanceId
     * @return task instance list
     */
    public List<TaskInstance> findPreviousTaskListByWorkProcessId(String processInstanceId) {
        return taskInstanceMapper.findValidTaskListByProcessId(processInstanceId, Flag.NO.getCode());
    }

    /**
     * update work process instance map
     *
     * @param processInstanceMap processInstanceMap
     * @return update process instance result
     */
    public int updateWorkProcessInstanceMap(ProcessInstanceMap processInstanceMap) {
        return processInstanceMapMapper.updateByPrimaryKey(processInstanceMap);
    }

    /**
     * create work process instance map
     *
     * @param processInstanceMap processInstanceMap
     * @return create process instance result
     */
    public int createWorkProcessInstanceMap(ProcessInstanceMap processInstanceMap) {
        int count = 0;
        if (processInstanceMap != null) {
            return processInstanceMapMapper.insert(processInstanceMap);
        }
        return count;
    }

    /**
     * find work process map by parent process id and parent task id.
     *
     * @param parentWorkProcessId parentWorkProcessId
     * @param parentTaskId        parentTaskId
     * @return process instance map
     */
    public ProcessInstanceMap findWorkProcessMapByParent(
            String parentWorkProcessId, String parentTaskId) {
        return processInstanceMapMapper.queryByParentId(parentWorkProcessId, parentTaskId);
    }

    /**
     * delete work process map by parent process id
     *
     * @param parentWorkProcessId parentWorkProcessId
     * @return delete process map result
     */
    public int deleteWorkProcessMapByParentId(String parentWorkProcessId) {
        return processInstanceMapMapper.deleteByParentProcessId(parentWorkProcessId);
    }

    /**
     * find sub process instance
     *
     * @param parentProcessId parentProcessId
     * @param parentTaskId    parentTaskId
     * @return process instance
     */
    public ProcessInstance findSubProcessInstance(String parentProcessId, String parentTaskId) {
        ProcessInstance processInstance = null;
        ProcessInstanceMap processInstanceMap =
                processInstanceMapMapper.queryByParentId(parentProcessId, parentTaskId);
        if (processInstanceMap == null
                || StringUtils.isEmpty(processInstanceMap.getProcessInstanceId())) {
            return processInstance;
        }
        processInstance = findProcessInstanceById(processInstanceMap.getProcessInstanceId());
        return processInstance;
    }

    /**
     * find parent process instance
     *
     * @param subProcessId subProcessId
     * @return process instance
     */
    public ProcessInstance findParentProcessInstance(String subProcessId) {
        ProcessInstance processInstance = null;
        ProcessInstanceMap processInstanceMap =
                processInstanceMapMapper.queryBySubProcessId(subProcessId);
        if (processInstanceMap == null
                || StringUtils.isEmpty(processInstanceMap.getProcessInstanceId())) {
            return processInstance;
        }
        processInstance = findProcessInstanceById(processInstanceMap.getParentProcessInstanceId());
        return processInstance;
    }

    /**
     * change task state
     *
     * @param state       state
     * @param startTime   startTime
     * @param host        host
     * @param executePath executePath
     * @param logPath     logPath
     * @param taskInstId  taskInstId
     */
    public void changeTaskState(
            Integer state,
            Date startTime,
            String host,
            String executePath,
            String logPath,
            String taskInstId) {
        TaskInstance taskInstance = taskInstanceMapper.selectByPrimaryKey(taskInstId);
        taskInstance.setState(state);
        taskInstance.setStartTime(startTime);
        taskInstance.setHost(host);
        taskInstance.setExecutePath(executePath);
        taskInstance.setLogPath(logPath);
        saveTaskInstance(taskInstance);
        logger.info("开始更新任务状态：{}", taskInstance);
    }

    /**
     * update process instance
     *
     * @param processInstance processInstance
     * @return update process instance result
     */
    public int updateProcessInstance(ProcessInstance processInstance) {
        return processInstanceMapper.updateByPrimaryKey(processInstance);
    }

    public int updateProcessInstanceByPrimaryKeySelective(ProcessInstance processInstance) {
        return processInstanceMapper.updateByPrimaryKeySelective(processInstance);
    }

    /**
     * update the process instance
     *
     * @param processInstanceId processInstanceId
     * @param processJson       processJson
     * @param globalParams      globalParams
     * @param scheduleTime      scheduleTime
     * @param flag              flag
     * @param locations         locations
     * @param connects          connects
     * @return update process instance result
     */
    public int updateProcessInstance(
            String processInstanceId,
            String processJson,
            String globalParams,
            Date scheduleTime,
            Flag flag,
            String locations,
            String connects) {
        ProcessInstance processInstance = processInstanceMapper.queryDetailById(processInstanceId);
        if (processInstance != null) {
            processInstance.setProcessInstanceJson(processJson);
            processInstance.setGlobalParams(globalParams);
            processInstance.setScheduleTime(scheduleTime);
            processInstance.setLocations(locations);
            processInstance.setConnects(connects);
            return processInstanceMapper.updateByPrimaryKey(processInstance);
        }
        return 0;
    }

    /**
     * change task state
     *
     * @param state      state
     * @param endTime    endTime
     * @param taskInstId taskInstId
     */
    public void changeTaskState(
            Integer state, Date endTime, Integer processId, String appIds, String taskInstId) {
        TaskInstance taskInstance = taskInstanceMapper.selectByPrimaryKey(taskInstId);
        taskInstance.setPid(processId);
        taskInstance.setAppLink(appIds);
        taskInstance.setState(state);
        taskInstance.setEndTime(endTime);
        saveTaskInstance(taskInstance);
    }

    /**
     * query schedule by id
     *
     * @param id id
     * @return schedule
     */
    public Schedule querySchedule(String id) {
        return scheduleMapper.selectByPrimaryKey(id);
    }

    /**
     * query Schedule by processDefinitionId
     *
     * @param processDefinitionId processDefinitionId
     * @see Schedule
     */
    public List<Schedule> queryReleaseSchedulerListByProcessDefinitionId(String processDefinitionId) {
        return scheduleMapper.queryReleaseSchedulerListByProcessDefinitionId(processDefinitionId);
    }

    /**
     * query need failOver process instance
     *
     * @param host host
     * @return process instance list
     */
    public List<ProcessInstance> queryNeedFailOverProcessInstances(String host) {

        return processInstanceMapper.queryByHostAndStatus(host, stateArray);
    }

    /**
     * process need failOver process instance
     *
     * @param processInstance processInstance
     */
    @Transactional(rollbackFor = Exception.class)
    public void processNeedFailOverProcessInstances(ProcessInstance processInstance) {
        // 1 update processInstance host is null
        processInstance.setHost(Constants.NULL);
        processInstanceMapper.updateByPrimaryKey(processInstance);

        // 2 insert into recover command
        Command cmd = new Command();
        cmd.setId(ComUtil.getId());
        cmd.setProcessDefinitionId(processInstance.getProcessDefinitionId());
        cmd.setCommandParam(
                String.format(
                        "{\"%s\":\"%s\"}",
                        Constants.CMDPARAM_RECOVER_PROCESS_ID_STRING, processInstance.getId()));
        cmd.setExecutorId(processInstance.getExecutorId());
        cmd.setCommandType(CommandType.RECOVER_TOLERANCE_FAULT_PROCESS.getCode());
        createCommand(cmd);
    }

    /**
     * query all need failOver task instances by host
     *
     * @param host host
     * @return task instance list
     */
    public List<TaskInstance> queryNeedFailOverTaskInstances(String host) {
        return taskInstanceMapper.queryByHostAndStatus(host, stateArray);
    }

    /**
     * find data source by id
     *
     * @param id id
     * @return datasource
     */
    public DataSourceDTO findDataSourceById(String id) {
        return dataSourceService.selectByPrimaryKey(id);
    }

    /**
     * update process instance state by id
     *
     * @param processInstanceId processInstanceId
     * @param executionStatus   executionStatus
     * @return update process result
     */
    public int updateProcessInstanceState(Integer processInstanceId, Integer executionStatus) {
        ProcessInstance instance = processInstanceMapper.selectByPrimaryKey(processInstanceId);
        instance.setState(executionStatus);
        return processInstanceMapper.updateByPrimaryKey(instance);
    }

    /**
     * find process instance by the task id
     *
     * @param taskId taskId
     * @return process instance
     */
    public ProcessInstance findProcessInstanceByTaskId(String taskId) {
        TaskInstance taskInstance = taskInstanceMapper.selectByPrimaryKey(taskId);
        if (taskInstance != null) {
            return processInstanceMapper.selectByPrimaryKey(taskInstance.getProcessInstanceId());
        }
        return null;
    }

    /**
     * find udf function list by id list string
     *
     * @param ids ids
     * @return udf function list
     */
    public List<UdfFunc> queryUdfFunListByIds(String[] ids) {
        return udfFuncMapper.queryUdfByIdStr(ids, null);
    }

    /**
     * find tenant code by resource name
     *
     * @param resName      resource name
     * @param resourceType resource type
     * @return tenant code
     */
    public String queryTenantCodeByResName(String resName, ResourceType resourceType) {
        return ConfigUtils.getInstance().getString("tentant", "root");
    }

    /**
     * find schedule list by process define id.
     *
     * @param ids ids
     * @return schedule list
     */
    public List<Schedule> selectAllByProcessDefineId(String[] ids) {
        return scheduleMapper.selectAllByProcessDefineArray(ids);
    }

    /**
     * get dependency cycle by work process define id and scheduler fire time
     *
     * @param masterId            masterId
     * @param processDefinitionId processDefinitionId
     * @param scheduledFireTime   the time the task schedule is expected to trigger
     * @return CycleDependency
     * @throws Exception if error throws Exception
     */
    public CycleDependency getCycleDependency(
            String masterId, String processDefinitionId, Date scheduledFireTime) throws Exception {
        List<CycleDependency> list =
                getCycleDependencies(masterId, new String[]{processDefinitionId}, scheduledFireTime);
        return list.size() > 0 ? list.get(0) : null;
    }

    /**
     * get dependency cycle list by work process define id list and scheduler fire time
     *
     * @param masterId          masterId
     * @param ids               ids
     * @param scheduledFireTime the time the task schedule is expected to trigger
     * @return CycleDependency list
     * @throws Exception if error throws Exception
     */
    public List<CycleDependency> getCycleDependencies(
            String masterId, String[] ids, Date scheduledFireTime) throws Exception {
        List<CycleDependency> cycleDependencyList = new ArrayList<>();
        if (ArrayUtils.isEmpty(ids)) {
            logger.warn("ids[] is empty!is invalid!");
            return cycleDependencyList;
        }
        if (scheduledFireTime == null) {
            logger.warn("scheduledFireTime is null!is invalid!");
            return cycleDependencyList;
        }

        String strCrontab = "";
        CronExpression depCronExpression;
        Cron depCron;
        List<Date> list;
        List<Schedule> schedules = this.selectAllByProcessDefineId(ids);
        // for all scheduling information
        for (Schedule depSchedule : schedules) {
            strCrontab = depSchedule.getCrontab();
            depCronExpression = CronUtils.parse2CronExpression(strCrontab);
            depCron = CronUtils.parse2Cron(strCrontab);
            CycleEnum cycleEnum = CronUtils.getMiniCycle(depCron);
            if (cycleEnum == null) {
                logger.error("{} is not valid", strCrontab);
                continue;
            }
            Calendar calendar = Calendar.getInstance();
            switch (cycleEnum) {
          /*case MINUTE:
          calendar.add(Calendar.MINUTE,-61);*/
                case HOUR:
                    calendar.add(Calendar.HOUR, -25);
                    break;
                case DAY:
                    calendar.add(Calendar.DATE, -32);
                    break;
                case WEEK:
                    calendar.add(Calendar.DATE, -32);
                    break;
                case MONTH:
                    calendar.add(Calendar.MONTH, -13);
                    break;
                default:
                    logger.warn(
                            "Dependent process definition's  cycleEnum is {},not support!!", cycleEnum.name());
                    continue;
            }
            Date start = calendar.getTime();

            if (depSchedule.getProcessDefinitionId().equals(masterId)) {
                list = CronUtils.getSelfFireDateList(start, scheduledFireTime, depCronExpression);
            } else {
                list = CronUtils.getFireDateList(start, scheduledFireTime, depCronExpression);
            }
            if (list.size() >= 1) {
                start = list.get(list.size() - 1);
                CycleDependency dependency =
                        new CycleDependency(
                                depSchedule.getProcessDefinitionId(),
                                start,
                                CronUtils.getExpirationTime(start, cycleEnum),
                                cycleEnum);
                cycleDependencyList.add(dependency);
            }
        }
        return cycleDependencyList;
    }

    /**
     * find last scheduler process instance in the date interval
     *
     * @param definitionId definitionId
     * @param dateInterval dateInterval
     * @return process instance
     */
    public ProcessInstance findLastSchedulerProcessInterval(
            String definitionId, DateInterval dateInterval) {
        return processInstanceMapper.queryLastSchedulerProcess(
                definitionId, dateInterval.getStartTime(), dateInterval.getEndTime());
    }

    /**
     * find last manual process instance interval
     *
     * @param definitionId process definition id
     * @param dateInterval dateInterval
     * @return process instance
     */
    public ProcessInstance findLastManualProcessInterval(
            String definitionId, DateInterval dateInterval) {
        return processInstanceMapper.queryLastManualProcess(
                definitionId, dateInterval.getStartTime(), dateInterval.getEndTime());
    }

    /**
     * find last running process instance
     *
     * @param definitionId process definition id
     * @param startTime    start time
     * @param endTime      end time
     * @return process instance
     */
    public ProcessInstance findLastRunningProcess(String definitionId, Date startTime, Date endTime) {
        return processInstanceMapper.queryLastRunningProcess(
                definitionId, startTime, endTime, stateArray);
    }

    /**
     * query user queue by process instance id
     *
     * @param processInstanceId processInstanceId
     * @return queue
     */
    public String queryUserQueueByProcessInstanceId(String processInstanceId) {
        String queue = "";
        ProcessInstance processInstance = processInstanceMapper.selectByPrimaryKey(processInstanceId);
        if (processInstance != null) {
            return processInstance.getQueue();
        }
        return queue;
    }

    /**
     * get task worker group
     *
     * @param taskInstance taskInstance
     * @return workerGroupId
     */
    public String getTaskWorkerGroup(TaskInstance taskInstance) {
        String workerGroup = taskInstance.getWorkerGroup();

        if (StringUtils.isNotBlank(workerGroup)) {
            return workerGroup;
        }
        String processInstanceId = taskInstance.getProcessInstanceId();
        ProcessInstance processInstance = findProcessInstanceById(processInstanceId);

        if (processInstance != null) {
            return processInstance.getWorkerGroup();
        }
        logger.info("task : {} will use default worker group", taskInstance.getId());
        return Constants.DEFAULT_WORKER_GROUP;
    }

    /**
     * get have perm project list
     *
     * @param userId userId
     * @return project list
     */
    public List<Project> getProjectListHavePerm(int userId) {
        List<Project> createProjects = projectMapper.queryProjectCreatedByUser(userId);
        List<Project> authedProjects = projectMapper.queryAuthedProjectListByUserId(userId);

        if (createProjects == null) {
            createProjects = new ArrayList<>();
        }

        if (authedProjects != null) {
            createProjects.addAll(authedProjects);
        }
        return createProjects;
    }

    /**
     * get have perm project ids
     *
     * @param userId userId
     * @return project ids
     */
    public List<String> getProjectIdListHavePerm(int userId) {
        List<String> projectIdList = new ArrayList<>();
        for (Project project : getProjectListHavePerm(userId)) {
            projectIdList.add(project.getId());
        }
        return projectIdList;
    }

    /**
     * list unauthorized udf function
     *
     * @param userId     user id
     * @param needChecks data source id array
     * @return unauthorized udf function list
     */
    public List<String> listUnauthorized(
            String userId, String[] needChecks, AuthorizationType authorizationType) {
        List<String> resultList = new ArrayList<>();

        if (!ArrayUtils.isEmpty(needChecks)) {
            Set<String> originResSet = new HashSet<>(Arrays.asList(needChecks));

            switch (authorizationType) {
                case UDF_FILE:
                case RESOURCE_FILE_ID:
                    Set<String> authorizedResourceFiles =
                            resourceMapper.listAuthorizedResourceById(userId, needChecks).stream()
                                    .map(Resource::getId)
                                    .collect(toSet());
                    originResSet.removeAll(authorizedResourceFiles);
                    break;
                case RESOURCE_FILE_NAME:
                    Set<String> authorizedResources =
                            resourceMapper.listAuthorizedResource(userId, needChecks).stream()
                                    .map(Resource::getFullName)
                                    .collect(toSet());
                    originResSet.removeAll(authorizedResources);
                    break;
                case DATASOURCE:
                    // TODO 需要增加数据源权限
                    // Set<String> authorizedDatasources = dataSourceService.listAuthorizedDataSource(userId,
                    // needChecks).stream().map(DataSource::getId).collect(toSet());
                    // originResSet.removeAll(authorizedDatasources);
                    break;
                case UDF:
                    Set<String> authorizedUdfs =
                            udfFuncMapper.listAuthorizedUdfFunc(userId, needChecks).stream()
                                    .map(UdfFunc::getId)
                                    .collect(toSet());
                    originResSet.removeAll(authorizedUdfs);
                    break;
            }

            resultList.addAll(originResSet);
        }

        return resultList;
    }

    /**
     * get resource by resource id
     *
     * @param resourceId resource id
     * @return Resource
     */
    public Resource getResourceById(String resourceId) {
        return resourceMapper.selectByPrimaryKey(resourceId);
    }

    /**
     * list resources by ids
     *
     * @param resIds resIds
     * @return resource list
     */
    public List<Resource> listResourceByIds(String[] resIds) {
        return resourceMapper.listResourceByIds(resIds);
    }

    /**
     * format task app id in task instance
     *
     * @param taskInstance 任务实例
     * @return format task app id
     */
    public String formatTaskAppId(TaskInstance taskInstance) {
        ProcessDefinition definition =
                this.findProcessDefineById(taskInstance.getProcessDefinitionId());
        ProcessInstance processInstanceById =
                this.findProcessInstanceById(taskInstance.getProcessInstanceId());
        if (definition == null || processInstanceById == null) {
            return "";
        }
        return String.format(
                "%s_%s_%s", definition.getId(), processInstanceById.getId(), taskInstance.getId());
    }

    /**
     * 更新流程定义的附加参数
     */
    public void updateProcessDefinitionExtraParams(String processDefinitionId, String extraParams) {
        processDefineMapper.updateProcessDefinitionExtraParams(processDefinitionId, extraParams);
    }

    /**
     * 获取通用节点信息
     *
     * @param nodeId nodeId
     * @return 通用节点信息
     */
    public TaskNodes getTaskNodesById(String nodeId) {
        return taskNodesService.selectByPrimaryKey(nodeId);
    }

    /**
     * 获取算法节点信息
     *
     * @param nodeId nodeId
     * @return 算法节点信息
     */
    public AlgoNodes getTaskAlgoNodesById(String nodeId) {
        return algoNodesService.selectByPrimaryKey(nodeId);
    }

    /**
     * 获取数据源信息
     *
     * @param inputDatasourceId 数据源Id
     * @return 数据源信息
     */
    public BaseDataSource getDataSource(String inputDatasourceId) {
        DataSourceDTO datasource = dataSourceService.queryDataSourceById(inputDatasourceId);
        return DataSourceFactory.getDatasource(
                datasource.getDatasourceType(), datasource.getConnectionParams());
    }

    /**
     * 获取数据源id
     *
     * @param tableId 表Id
     * @return 数据源id
     */
    public String getDataSourceIdByTableId(String tableId) {
        return projectTableService.getDatasourceIdByTableId(tableId);
    }

    public String getTableNameByTableId(String tableId) {
        return projectTableService.getTableNameByTableId(tableId);
    }

    public void updateDCSTaskStatus(TaskInstance taskInstance) {
        // TODO 任务执行结果通知到DCS采集系统
        try {
            String processDefinitionId = taskInstance.getProcessDefinitionId();
            ProcessDefinition processDefinition =
                    processDefineMapper.queryByDefineId(processDefinitionId);
            // 迁移任务类型并且是DATAX节点
            if (processDefinition != null
                    && (!processDefinition.getType().equals(ProcessConstants.JOB_DATAPROCESS))) {
                if (TaskType.DATAX.name().equals(taskInstance.getTaskType())) {
                    Boolean result =
                            taskAPI.updateTaskStatus(dcsAPIUrl, taskInstance.getId(), taskInstance.getState());
                    logger.info("update taskInstance exeState to DCS result {}", result);
                }
            }
        } catch (Exception e) {
            logger.error("update taskInstance exeState to DCS error {}", e.getMessage());
        }
    }

    /**
     * @param resourceId resource id
     * @return PythonResourceDTO
     */
    public PythonResourceDTO getPythonResourceById(String resourceId) {
        return resourceMapper.getPythonResourceById(resourceId);
    }

    public void updatePythonResourceById(String resourceId, Integer status) {
        resourceMapper.updatePythonResourceById(resourceId, status);
    }

    public ProcessDefinition getProcessDefinitionByProcessInstanceId(String processInstanceId) {
        return this.processDefineMapper.getDefinitionByProcessInstanceId(processInstanceId);
    }

    public String getBaseResourceFullName(String projectId) {
        StorageResourceConf storageResourceConf =
                fileOperateCommonService.getStorageResourceConf(projectId);

        String baseFilePath = fileOperateCommonService.getDDSStoragePath(storageResourceConf);

        return baseFilePath + "/" + projectId + "/file";
    }

    /**
     * 重新计算、暂停等操作
     *
     * @param processDefineId
     * @param processInstanceId
     * @param type
     */
    public void execute(String processDefineId, String processInstanceId, ExecuteType type) {
        this.execService.execute(processDefineId, processInstanceId, type);
    }

    /**
     * 更新Resource
     *
     * @param resource
     */
    public void updateResource(Resource resource) {
        this.resourceMapper.updateByPrimaryKeySelective(resource);
    }

    /**
     * 保存yarn的appid
     */
    public void updateTaskAppIdByProcessInstanceId(String processInstanceId, String appId) {
        this.taskInstanceMapper.updateTaskAppIdByProcessInstanceId(processInstanceId, appId);
    }

    /**
     * 获取资源对象
     */
    public BdpPrmEngineResourceDTO findBdpPrmEngineResourceById(String resourceId) {
        return this.bdpPrmEngineResourceApi.getEngineResourceById(resourceId);
    }

    /**
     * 更新数据质量任务状态
     *
     * @param taskInstance
     */
    public void updateDgsQualityTaskStatus(TaskInstance taskInstance) {
        String processDefinitionId = taskInstance.getProcessDefinitionId();
        ProcessDefinition processDefinition = processDefineMapper.queryByDefineId(processDefinitionId);
        if (processDefinition == null) {
            logger.warn("未查询到指定流程定义[{}]", processDefinitionId);
            return;
        }

        String projectId = processDefinition.getProjectId();
        try {
            // 数据质量专属项目空间，系统初始化时创建
            if (projectId != null && projectId.equals("777777")) {
                if (taskInstance.getState() == 7) {
                    // DDS任务状态；7：成功；
                    // 质量任务状态：0：草稿；1：运行中；6：失败；7：成功
                    this.qualityJobApi.updateQaJobStatus(processDefinitionId, taskInstance.getProcessInstanceId(), taskInstance.getState());
                } else {
                    // 如果DDS任务状态未成功，统一认为任务失败
                    this.qualityJobApi.updateQaJobStatus(processDefinitionId, taskInstance.getProcessInstanceId(), 6);
                }

                logger.info("更新质量任务状态到DGS成功,procDefeId={},taskId={},taskName={},status={}",
                        processDefinitionId, taskInstance.getId(), processDefinition.getName(), taskInstance.getState());
            }
        } catch (Exception e) {
            logger.error("更新质量任务状态到DGS成功,procDefeId={},taskId={},taskName={},status={},errMsg={}",
                    processDefinitionId, taskInstance.getId(), processDefinition.getName(), taskInstance.getState(),
                    e.getCause());
        }
    }
}
